package com.chukun.flink.stream.window.source;

import com.chukun.flink.stream.util.TimeUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口数据准备
 * @createTime 2022年05月22日 21:48:00
 */
public class SourceForWindow implements SourceFunction<Tuple3<String,Integer, String>> {

    private volatile boolean isRunning = true;

    // 发送元素的间隔时间
    private long sleepTime;

    // 标识元素是否按照sleepTime为间隔发送
    private boolean stopSession = false;

    public static final String[] words = new String[] {
            "flink","flink","flink",
            "java","hadoop","spark",
            "java","spark","doris"
    };

    public SourceForWindow(long sleepTime, boolean stopSession) {
        this.sleepTime = sleepTime;
        this.stopSession = stopSession;
    }

    @Override
    public void run(SourceContext<Tuple3<String, Integer, String>> ctx) throws Exception {
        int count = 0;
        while (isRunning) {
            // 遍历数组中的元素
            String word = words[count % words.length];
            String time = TimeUtils.getHHmmss(System.currentTimeMillis());
            Tuple3<String, Integer, String> tuple = Tuple3.of(word, count, time);
            // 发送元素
            ctx.collect(tuple);
            System.out.println("send data : " +tuple);

            if (stopSession && count == words.length) {
                TimeUnit.SECONDS.sleep(10);
            } else {
                Thread.sleep(sleepTime);
            }
            count++;
        }
    }

    @Override
    public void cancel() {
       this.isRunning = false;
    }
}
